package com.atguigu.edu.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.utils.DimUtil;
import com.atguigu.edu.realtime.utils.TreadPoolUtil;
import org.apache.calcite.rel.core.Join;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author 杨晨昱
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> implements DimJoinFunction<T> {
    private ExecutorService executorService;
    private String tableName;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        executorService = TreadPoolUtil.getInstance();
    }

    public void asyncInvoke(T obj, ResultFuture<T> resultFuture) {
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                String key = getkey(obj);
                JSONObject dimInfoJsonObj = DimUtil.getDimInfo(tableName, key);
                if (dimInfoJsonObj != null) {
                    join(obj, dimInfoJsonObj);
                }
                resultFuture.complete(Collections.singleton(obj));
            }
        });
    }
}
